package com.wudl.flink.stream.tablesql

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.DataTypes
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema}
import org.apache.flink.types.Row

object FileTableOutinput {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val tableEnv = StreamTableEnvironment.create(env)
    val filePath = "F:\\ideaWorkSpace2020\\demo\\Flink-wudl\\src\\main\\resources\\sensor.txt";
    tableEnv.connect(new FileSystem().path(filePath))
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("tmp", DataTypes.DOUBLE())
      ).createTemporaryTable("kafkainputTable")

    val fileTable = tableEnv.from("kafkainputTable")
    // 转化
    val resultTable = fileTable.select('id,'tmp)
      .filter('id === "sensor_1")

    // 3.2 聚合转换
    val aggTable = fileTable
      .groupBy('id) // 基于id分组
      .select('id, 'id.count as 'count)

    //  输出到文件
    val outputPath = "F:\\ideaWorkSpace2020\\demo\\Flink-wudl\\src\\main\\resources\\output.txt"
    tableEnv.connect(new FileSystem().path(outputPath))
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("tmperature", DataTypes.DOUBLE()))
      .createTemporaryTable("outputTable")
    resultTable.insertInto("outputTable")
    resultTable.toAppendStream[(String, Double)].print("result")
    aggTable.toRetractStream[Row].print("agg")

    env.execute("outputTable")

  }
}
